/*
    异步工作器 - 创建异步工作线程来完成日志的具体落地
    采用双缓冲区的思想，来控制业务线程和异步线程的生产与消费关系
*/

#pragma once
#include "buffer.hpp"
#include <mutex>
#include <condition_variable>
#include <thread>
#include <functional>
#include <atomic>

namespace logging
{
    using func_t = std::function<void(buffer &buf)>;
    enum class asyncType
    {
        ASYNC_SAFE,
        ASYNC_UNSAFE
    };
    
    class asyncLooper
    {
    public:
        using s_ptr = std::shared_ptr<asyncLooper>;

    public:
        asyncLooper(func_t callback, asyncType looper_type = asyncType::ASYNC_SAFE)
            : _isRunning(false),
              _thread(std::thread(&asyncLooper::threadRoutine, this)),
              _callBack(callback),
              _async_type(looper_type)
        {}
        ~asyncLooper() { stop(); }
        void stop()
        {
            // 当异步工作器要关闭时，要将所有的线程唤醒
            _isRunning = false;
            _consume_cond.notify_all();
            _thread.join();
        }
        void push(const char *data, size_t len)
        {
            // isRunning是原子类型，所以不用担心线程安全问题
            if(!_isRunning) return;
            // 1.申请锁
            std::unique_lock<std::mutex> lock(_mutex);
            // 2.判断输入缓冲区还有没有足够的空间，如果不够，则需要让业务线程阻塞
            if(_async_type == asyncType::ASYNC_SAFE)
                _product_cond.wait(lock, [&](){ return _product_buf.writeAbleSize() >= len;/*输入缓冲区可写大小，大小len*/});
            // 3.将数据写入到输入缓冲区中
            _product_buf.push(data, len);
            // 到这里，就将锁释放掉，处理数据不需要占有锁
            // 此时已经有了数据，就可以唤醒消费者了
            _consume_cond.notify_one();
        }

    private:
        void threadRoutine()
        {
            _isRunning = true;
            while(1) {
                // 给锁一个声明周期，在交换完缓冲区之后，就释放掉锁，处理数据不需要加锁
                {
                    // 1.加锁，保证异步线程进行缓冲区交换时是线程安全的
                    std::unique_lock<std::mutex> lock(_mutex);
                    if(!_isRunning && _product_buf.empty()) break;
                    // 2.如果输入缓冲区有数据，或者异步线程要退出了，才能继续运行，否则就得阻塞等待
                    //   输出缓冲区默认为空，所以，只要输入缓冲区有消息就交换
                    //   下面的处理过程，会将输出缓冲区处理完，重新置空
                    _consume_cond.wait(lock, [&](){ return !_isRunning || !_product_buf.empty(); });
                    _consume_buf.swap(_product_buf);
                    // 3.唤醒生产者线程
                    if(_async_type == asyncType::ASYNC_SAFE)
                        _product_cond.notify_all();
                }
                // 4.处理输出缓冲区
                _callBack(_consume_buf);
                // 5.将输出缓冲区置空
                _consume_buf.reset();
            }
        }
        func_t _callBack; // 回调函数
    private:
        // 双缓冲区
        buffer _product_buf; // 生产者缓冲区，用于业务线程push格式字符串
        buffer _consume_buf; // 消费者缓冲区，用于异步线程读取字符串，进行落地
        std::mutex _mutex;   // 互斥锁
        // 条件变量
        std::condition_variable _product_cond; // 生产者条件变量，当条件不满足时，让生产者线程阻塞
        std::condition_variable _consume_cond; // 消费者条件变量，当条件不满足时，让异步线程阻塞
        std::atomic<bool> _isRunning;
        std::thread _thread;
        asyncType _async_type;
    };
}